package com.ruyuan.starter.async;

import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.dsl.Disruptor;
import com.ruyuan.starter.async.autoconfigure.BossConfig;
import com.ruyuan.starter.async.autoconfigure.WorkerConfig;
import com.ruyuan.starter.async.event.BaseEvent;
import com.ruyuan.starter.async.event.BossEvent;
import com.ruyuan.starter.async.handler.BossEventHandler;
import com.ruyuan.starter.async.thread.NamedDaemonThreadFactory;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author zhonghuashishan
 */
public class BossEventBus {

    private final Disruptor<BossEvent> bossRingBuffer;

    public BossEventBus(BossConfig bossConfig,
                        WorkerConfig workerConfig) {

        // 1、先把WorkEventBus准备好
        WorkEventBusManager workEventBusManager = WorkEventBusManager.getSingleton();
        for (WorkerConfig.Config config : workerConfig.getWorkers()) {
            workEventBusManager.register(config);
        }
        // 2、再把BossEventBus准备好
        bossRingBuffer = new Disruptor<>(BossEvent::new, bossConfig.getRingbufferSize(),
                NamedDaemonThreadFactory.getInstance("BossEventBus"));
        BossEventHandler[] eventHandlers = new BossEventHandler[bossConfig.getEventHandlerNum()];
        for (int i = 0; i < eventHandlers.length; i++) {
            eventHandlers[i] = new BossEventHandler();
        }
        bossRingBuffer.handleEventsWithWorkerPool(eventHandlers);
        bossRingBuffer.start();
    }

    public boolean publish(String channel, BaseEvent event, AsyncContext context) {
        EventTranslator<BossEvent> translator = (e, s) -> {
            e.channel = channel;
            e.event = event;
            e.context = context;
        };
        boolean success = bossRingBuffer.getRingBuffer().tryPublishEvent(translator);
        if (!success) {
            System.out.println("bossRingbuffer满了");
        }
        return success;
    }

}